﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;

namespace Cluster
{
    public abstract partial class ClusterManager
    {
        /// <summary>
        /// Replica meta-data
        /// </summary>
        [Serializable]
        protected class Replica
        {
            [NonSerialized]
            public string ZkPath;                                   // name iz /replication tree

            public string ServerName;                               // the replica name

            public string ReplicatingNode;                          // the name of the replicating node
        }

        protected List<Replica> ReplicationView { get; set; }       // local replicas view of the cluster

        private PathWatcher replicationWatcher;                     // watches changes in the replication path

        private ReplicationQueue replicationQueue;                  // distributed prioritized tasks queue

        private bool replicationScheduled;                          // was a replication scheduled but not executed yet?

        private bool replicatedSelf;                                // did we already replicate ourself?

        /// <summary>
        /// Performs a replication requst on the remote node
        /// </summary>
        /// <param name="name">the server name we want to replicate</param>
        /// <param name="node">the node we send the replica to</param>
        protected abstract void RequestReplicationFromRemoteNode(string name, ClusterNode node);

        /// <summary>
        /// Discards outdated replicas
        /// </summary>
        /// <param name="name">the server name we want to discard</param>
        /// <param name="node">the node we're asking to discard the server</param>
        protected abstract void RequestDiscardReplicationFromRemoteNode(string name, ClusterNode node);

        /// <summary>
        /// Updates the local view of the replications from the Zoo Keeper server.
        /// </summary>
        [MethodImpl(MethodImplOptions.Synchronized)]
        private void UpdateReplicationView()
        {
            Debug("Updating replication view");

            var view = new List<Replica>();

            while (true) //  the nodes might dissappear while retrieving their data so we need to retry
            {
                view.Clear();

                var replicas = GetZkChildren(ZkReplicationPath);

                try
                {
                    // get meta-data for each replica
                    foreach (var name in replicas)
                    {
                        var replica = Deserialize<Replica>(GetZkNodeData(ZkReplicationPath + "/" + name));
                        replica.ZkPath = name;

                        view.Add(replica);
                    }

                    break;
                }
                catch (ClusterException)
                {
                    Debug("Replication view was changed during node data retrival, retrying");
                }
            }

            Debug("Retrieved replication view ({0} servers)", view.Count);

            // print the retrieved view to log - but only if leader (to avoid huge logs)
            if (isLeader)
            {
                var sb = new StringBuilder();
                sb.AppendLine("Replication view:");

                foreach (var replica in view)
                {
                    sb.AppendLine(string.Format("\t{0}\t name={1}\t replacting node={2}",
                        replica.ZkPath, replica.ServerName, replica.ReplicatingNode));
                }

                Debug(sb.ToString());
            }

            ReplicationView = view;
        }

        /// <summary>
        /// Setups the tasks queue and replication path watcher
        /// </summary>
        private void SetupReplication()
        {
            replicationQueue = new ReplicationQueue(this);

            replicationQueue.TaskDequeued += ((o, e) =>
            {
                switch (e.Type)
                {
                    case ReplicationQueue.TaskType.Replicate:
                        DoReplication();
                        break;
                    case ReplicationQueue.TaskType.Balance:
                        DoBalancing();
                        break;
                }
            });
        }

        /// <summary>
        /// Checks if replication is required.
        /// If yes, schedules a rreplication task - a new entry is added to the distributed
        /// prioritized tasks queue and executed only when in the top of this queue 
        /// </summary>
        private void ScheduleReplication()
        {
            Debug("Checking if need to replicate");

            // check if we need to schedule replication - replication was not already scheduled and we
            // have a server that is not replicated on other nodes
            if (!replicationScheduled && GetServerNameToReplicate() != null)
            {
                try
                {
                    replicationScheduled = true;
                    replicationQueue.RegisterTask(ReplicationQueue.TaskType.Replicate);

                    Debug("Scheduling replication");
                }
                catch (Exception)
                {
                    replicationScheduled = false;
                    throw;
                }
            }
            else
            {
                Debug("No replication required");
            }
        }

        /// <summary>
        /// Perform the actual replication
        /// </summary>
        private void DoReplication()
        {
            Debug("Starting replication method");

            replicationScheduled = false;

            // get a server name that we need to replicate
            var name = GetServerNameToReplicate();
            if (name == null)
            {
                Debug("Did not find a server to replicae");
                return;
            }            

            // find a node that we can replicate to
            // balancing would be performed later
            var cn = ChoseReplicationNode();
            if (cn == null)
            {
                Debug("No other server found in cluster, can't replicate");
                return;
            }
            
            try
            {
                // replicate the chosen server to the chosen node
                ReplicateServerToNode(name, cn);
            }
            catch (ClusterException)
            {
                // we can't handle cluster exceptions here but we do want to
                // "handle" any remote server exception
                throw;
            }
            catch (Exception)
            {
                // let's try again
                // before the next scheduling task runs, we'll get an updated view
                // of the cluster and some dead nodes might go away
                ScheduleReplication();
            }

            // we might need to replicate more but let's do it in the next time-slice
            // a new replication will be scheduled after OnReplicationTreeChange is called (it
            // will be called if we succesfully replicated using ReplicateServerToNode)
        }

        /// <summary>
        /// Performs replication of the specified server to the specified node
        /// </summary>
        /// <param name="name">the server name to replicate</param>
        /// <param name="cn">the node we're replicating to</param>
        private void ReplicateServerToNode(string name, ClusterNode cn)
        {
            try
            {
                Debug("Trying to replicate {0} on {1} node ({2} server)", name, cn.ZkPath, cn.Name);
            
                // handle replication request in derived classes
                RequestReplicationFromRemoteNode(name, cn);
                
                Info("Replicated {0} on {1} node ({2} server)", name, cn.ZkPath, cn.Name);
            
                if (name == ServerName)
                {
                    replicatedSelf = true;
                }

                cn.ReplicasNumber++; // temporary, for local calculations
            }
            catch (ClusterException e)
            {
                WarnEx(string.Format("Failed to replicate self on {0} node (ZK error)", cn.ZkPath), e);
                throw;
            }
            catch (Exception e)
            {
                WarnEx(string.Format("Failed to replicate self on {0} node (node communication exception)", cn.ZkPath), e);
                throw;
            }
        }

        /// <summary>
        /// Removes outdated self replicas
        /// </summary>
        private void DiscardOldReplicas()
        {
            // get the current cluster and replication views
            UpdateClusterView();
            UpdateReplicationView();

            // get self replicas on other nodes
            var res = ReplicationView.Where(replica => replica.ServerName == ServerName);

            // ask these replicas to discard their old replicas of us 
            foreach (var replica in res)
            {
                Info("Discrding outdated self replication from node {0} ({1} server)", replica.ZkPath, replica.ReplicatingNode);

                // get it's node
                var cn = ClusterView.FirstOrDefault(n => n.ZkPath == replica.ReplicatingNode);
                if (cn == null)
                {
                    Warn("Replica {0} points to invalid cluster node {1}", replica.ZkPath, replica.ReplicatingNode);
                    continue;
                }

                try
                {
                    // handle discard request in derived classes
                    RequestDiscardReplicationFromRemoteNode(ServerName, cn);
                }
                catch (Exception e)
                {
                    WarnEx("Failed to discard old replicas from node" + cn.ZkPath, e);
                }
            }

            ClusterView.Clear(); // we want to trigger ClusterViewChanged in the future
        }

        /// <summary>
        /// Adds a new replication entry to the Zoo Keeper server
        /// </summary>
        /// <param name="name">the server we're replicating that should be added</param>
        [MethodImpl(MethodImplOptions.Synchronized)]
        public void RegisterServerReplication(string name)
        {
            Debug("Registering server replication for " + name);

            // fill replica structure
            var replica = new Replica
            {
                ServerName = name,
                ReplicatingNode = SelfNode.ZkPath,
            };

            ReplicationView.Add(replica); // temporary, for local calculations

            // add to Zoo Keeper
            RegisterReplica(replica);

            // update local replicas number and save it to Zoo Keeper
            if (name != ServerName)
            {
                SelfNode.ReplicasNumber++;
                UpdateSelfNodeInCluster();
            }

            // check if we need to balance now
            ScheduleBalancing();
        }

        /// <summary>
        /// Removes a replication entry from the Zoo Keeper server
        /// </summary>
        /// <param name="name">the server we're removing</param>
        [MethodImpl(MethodImplOptions.Synchronized)]
        public void UnregisterReplicatedServerBySelf(string name)
        {
            Debug("Unregistering server replication for " + name);

            // get the replica struct
            var replica = ReplicationView.FirstOrDefault(r => r.ServerName == name && r.ReplicatingNode == SelfNode.ZkPath);
            if (replica == null || (replica.ReplicatingNode == SelfNode.ZkPath && replica.ServerName == ServerName))
            {
                throw new ArgumentException("name");
            }

            // remove from Zoo Keeper
            UnregisterReplica(replica);

            // update local replicas number and save it to Zoo Keeper
            SelfNode.ReplicasNumber--;
            UpdateSelfNodeInCluster();
        }

        /// <summary>
        /// Retrieves a server that should be replicated (we're the only replica in the cluster)
        /// </summary>
        /// <returns>null of no such server</returns>
        private string GetServerNameToReplicate()
        {
            Debug("Looking for server to replicate");

            // check if we backed-up ourself
            if (replicatedSelf == false)
            {
                Debug("Replication required (not replicated self yet)");
                return ServerName;
            }

            Debug(".. Self server is replicated");

            // check if the backup node is alive
            var replicatingNode = ReplicatingNodeName;
            if (string.IsNullOrEmpty(replicatingNode) || ClusterView.Any(cn => cn.ZkPath == replicatingNode) == false)
            {
                Debug("Replication required (the current backup node is down)");
                return ServerName;
            }

            Debug(".. Backup node is up");

            // check if we store servers that no one else replicates
            var hostedReplicas = HostedReplicas.ToList();
            var replica = hostedReplicas.FirstOrDefault(r =>
                ReplicationView.Any(rr =>
                    rr.ServerName == r.ServerName &&                                // has the same server name
                    rr.ReplicatingNode != SelfNode.ZkPath                           // replicated by some other node
                ) == false);                                                        // => no other node replicates this server

            if (replica != null)
            {
                // we're the only instance of some other server
                Debug("Replication required (only backup of other node {0})", replica.ServerName);

                return replica.ServerName;
            }

            Debug(".. Did not find a server to replicate");

            return null;
        }

        /// <summary>
        /// Choses the node we're replicating to
        /// </summary>
        /// <returns>null if no such node</returns>
        private ClusterNode ChoseReplicationNode()
        {
            var res = from cn in ClusterView
                      where cn != SelfNode // && cn.Name != SelfNode.Name
                      orderby cn.ReplicasNumber
                      select cn;

            return res.FirstOrDefault();
        }

        /// <summary>
        /// Create a new replica znode in the Zoo Keeper
        /// </summary>
        /// <param name="replica"></param>
        private void RegisterReplica(Replica replica)
        {
            var zpath = CreateZkNode(ZkReplicationPath + "/" + replica.ServerName, false, true, Serialize(replica));
            
            Debug("Registered replication {0} for server {1}", zpath, replica.ServerName);
        }

        /// <summary>
        /// Removes a replica znode form the Zoo Keeper
        /// </summary>
        /// <param name="replica"></param>
        private void UnregisterReplica(Replica replica)
        {
            RemoveZkPath(ZkReplicationPath + "/" + replica.ZkPath);
            
            Debug("Unregistered replication {0} for server {1}", replica.ZkPath, replica.ServerName);
        }
        
        /// <summary>
        /// Watches for changes in the replication Zoo Keeper path
        /// </summary>
        private void WatchReplicationPath()
        {
            if (replicationWatcher == null)
            {
                replicationWatcher = new PathWatcher(OnReplicationTreeChange, WatchReplicationPath);
            }

            WatchZkPath(ZkReplicationPath, replicationWatcher);
        }

        /// <summary>
        /// Called when a change is detected by Zoo Keeper in the replication path
        /// </summary>
        private void OnReplicationTreeChange()
        {
            Debug("Replication tree changed");

            UpdateReplicationView();
            UpdateClusterViewAndNotify(); // we need the new "replicates" values

            ScheduleReplication(); // check if need replicate/balance now (ClusterViewChanged might not be triggred)
            ScheduleBalancing();
        }

        /// <summary>
        /// The node name (ZK path) of out replicator
        /// </summary>
        protected string ReplicatingNodeName
        {
            get
            {
                var replica = ReplicationView.FirstOrDefault(r => r.ServerName == ServerName && r.ReplicatingNode != SelfNode.ZkPath);
                return replica != null ? replica.ReplicatingNode : null;
            }
        }

        /// <summary>
        /// The replicas saved in this node
        /// </summary>
        private IEnumerable<Replica> HostedReplicas
        {
            get { return ReplicationView.Where(r => r.ReplicatingNode == SelfNode.ZkPath); }
        }

        /// <summary>
        /// The Zoo Keeper replication path
        /// </summary>
        private string ZkReplicationPath
        {
            get { return "/" + ClusterName + "/replication"; }
        }
    }
}
